package org.egomsl.mw.region;


import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.egomsl.mw.Constants;
import org.egomsl.mw.record.Record;
import org.egomsl.mw.segment.ImmutableSegment;
import org.egomsl.mw.segment.MutableSegment;
import org.egomsl.mw.segment.SegmentBlockWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Pattern;

public class RegionWriter {
    private static Logger log = LoggerFactory.getLogger(RegionWriter.class);

    private boolean appendAllowed;
    private AtomicInteger seqNum;

    private Future<?> segmentWriterFuture;
    private MutableSegment active;

    private FileSystem fs;
    private Path regionDir;
    private FileMeta appendFile;

    private SegmentBlockWriter segmentBlockWriter;

    public RegionWriter(FileSystem fs, Path regionDir) throws IOException {
        this(fs, regionDir, Constants.MAX_BLOCK, Constants.BLOCK_SIZE);
    }

    public RegionWriter(FileSystem fs, Path regionDir, int maxBlock, int blockSize) throws IOException {
        this(fs, regionDir, maxBlock, blockSize, true);
    }

    public RegionWriter(FileSystem fs, Path regionDir,
                        int maxBlock, int blockSize,
                        boolean appendAllowed) throws IOException {
        this(fs, regionDir, maxBlock, blockSize, appendAllowed, 0);
    }

    public RegionWriter(FileSystem fs, Path regionDir,
                        int maxBlock, int blockSize,
                        boolean appendAllowed, int seqNum) throws IOException {
        this.fs = fs;
        this.regionDir = regionDir;
        this.segmentBlockWriter = new SegmentBlockWriter(fs);

        this.appendAllowed = appendAllowed;
        this.appendFile = null;
        this.seqNum = new AtomicInteger(seqNum);

        active = new MutableSegment(maxBlock, blockSize);

        init();
    }

    private void init() throws IOException {
        if(!fs.exists(regionDir)) {
            log.info("Making new region directory: " + regionDir);
            fs.mkdirs(regionDir);
            return ;
        }

        List<FileMeta> fileMetas = RegionUtils.getSegmentFiles(fs, regionDir);

        if(fileMetas.size() > 0) {
            FileMeta meta = fileMetas.get(0);

            if(meta.getSeqNum() >= seqNum.get()) {
                seqNum = new AtomicInteger(meta.getSeqNum() + 1);
            }

            String name = meta.getName();
            if(Pattern.matches(RegionConstants.sequentialBlockFilePattern, name)) {
                appendFile = meta;
            }
        }
    }

    public void add(Record record) {
        boolean shouldFlush = !active.add(record);
        if(shouldFlush) {
            if (segmentWriterFuture == null ||
                    segmentWriterFuture.isDone() ||
                    segmentWriterFuture.isCancelled()) {
                flush();
            }else {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    log.warn("Interrupted", e);
                }
            }
        }
    }

    public void syncFlush() {
        flush();
        try {
            segmentWriterFuture.get();
        }catch (Exception e) {
            log.error("Sync flush failed.", e);
        }
    }

    public void flush() {
        Path filePath = null;
        boolean append = false;

        ImmutableSegment segment = active.snapshot();
        if(segment.getTotalSize() == 0) {
            log.info("No data to flush for region: " + regionDir);
            return;
        }

        //default value
        int num = seqNum.addAndGet(1);
        String fileName = String.format(
                RegionConstants.reversedBlockFileFormat, num);
        filePath = new Path(regionDir, fileName);

        //two case for append
        if (appendAllowed) {
            int size = segment.getTotalSize();
            if(size <= RegionConstants.maxAppendFileSize) {
                segment.ensureOrder(false);

                if (appendFile != null &&
                        size + appendFile.getSize() < RegionConstants.maxAppendFileSize) {
                    append = true;
                    filePath = appendFile.getPath();
                    appendFile.setSize(size + appendFile.getSize());
                } else {
                    fileName = String.format(
                            RegionConstants.sequentialBlockFileFormat, num);
                    filePath = new Path(regionDir, fileName);

                    appendFile = new FileMeta(filePath, num, false);
                    appendFile.setSize(segment.getTotalSize());
                }
            }else {
                appendFile = null;
            }
        }

        segmentWriterFuture = segmentBlockWriter.flush(segment, filePath, append);
    }

    public Future<?> getSegmentWriterFuture() {
        return segmentWriterFuture;
    }

    public ImmutableSegment snapshot() {
        return active.snapshot();
    }

    public AtomicInteger getSeqNum() {
        return seqNum;
    }

}